package com.gxlevi.task

import com.gxlevi.bean.{ClickLogWide, Message}
import com.gxlevi.util.HBaseUtil
import org.apache.commons.lang.StringUtils
import org.apache.commons.lang.time.FastDateFormat
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.api.scala._

object PreprocessTask {
  def process(waterMarkDataStream: DataStream[Message]) = {
    val clickWideDataStream: DataStream[ClickLogWide] = waterMarkDataStream.map {
      msg =>
        val yearMonth = timestamp2Str(msg.timeStamp, "yyyyMM")
        val yearMonthDay = timestamp2Str(msg.timeStamp, "yyyyMMdd")
        val yearMonthDayHour = timestamp2Str(msg.timeStamp, "yyyyMMddHH")
        val address = msg.clickLog.country + msg.clickLog.province + msg.clickLog.city

        val isNewTuple: (Int, Int, Int, Int) = isNewProcess(msg)
        ClickLogWide(
          msg.clickLog.channelID,
          msg.clickLog.categoryID,
          msg.clickLog.produceID,
          msg.clickLog.country,
          msg.clickLog.province,
          msg.clickLog.city,
          msg.clickLog.network,
          msg.clickLog.source,
          msg.clickLog.browserType,
          msg.clickLog.entryTime,
          msg.clickLog.leaveTime,
          msg.clickLog.userID,
          msg.count,
          msg.timeStamp,
          address,
          yearMonth,
          yearMonthDay,
          yearMonthDayHour,
          isNewTuple._1,
          isNewTuple._2,
          isNewTuple._3,
          isNewTuple._4
        )
    }
    clickWideDataStream
  }

  private def isNewProcess(message: Message) = {
    var isNew = 0
    var isHourNew = 0
    var isDayNew = 0
    var isMonthNew = 0

    val tableName = "channel_user_history"
    val cfName = "info"
    val rowKey = message.clickLog.userID + ":" + message.clickLog.channelID
    val userIdColName = "userId"
    val channelIdColName = "channelId"
    val lastVisitTimeColName = "channelId"

    val userIdValue = HBaseUtil.getData(tableName, rowKey, cfName, userIdColName)

    if (StringUtils.isBlank(userIdValue)) {
      isNew = 1
      isDayNew = 1
      isMonthNew = 1
      isHourNew = 1
      HBaseUtil.putMapData(tableName, rowKey, cfName, Map(
        userIdColName -> message.clickLog.userID,
        channelIdColName -> message.clickLog.channelID,
        lastVisitTimeColName -> message.timeStamp.toString
      ))
    } else {
      isNew = 0
      val lastVisitedTime: String = HBaseUtil.getData(tableName, rowKey, cfName, lastVisitTimeColName)
      isHourNew = compareDate(message.timeStamp, lastVisitedTime.toLong, "yyyyMMddHH")
      isDayNew = compareDate(message.timeStamp, lastVisitedTime.toLong, "yyyyMMdd")
      isMonthNew = compareDate(message.timeStamp, lastVisitedTime.toLong, "yyyyMM")

      HBaseUtil.putData(tableName, rowKey, cfName, lastVisitTimeColName, message.timeStamp.toString)
    }
    (isNew, isHourNew, isDayNew, isMonthNew)
  }

  def timestamp2Str(timestamp: Long, format: String): String = {
    FastDateFormat.getInstance(format).format(timestamp)
  }

  def compareDate(currentTime: Long, historyTime: Long, format: String) = {
    var result = timestamp2Str(historyTime, format).compareTo(timestamp2Str(currentTime, format))
    if (result > 0) {
      result = 1
    } else {
      result = 0
    }
    result
  }

  def main(args: Array[String]): Unit = {
    println(compareDate(1558104383523L, 1568304383523L, "yyyyMMdd"))
  }
}
